﻿using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace CGF
{
    public class TCPClient : ComponentBase, IAwake<string, string>, IUpdate, IDestroy
    {
        class RPCTask
        {
            public CTask task;
            int protoId;
            Queue<RPCTask> taskQueue;

            public RPCTask(int protoId, CTask task, Queue<RPCTask> taskQueue, Type protoType)
            {
                this.protoId = protoId;
                this.task = task;
                this.taskQueue = taskQueue;
                ListenTimeout(protoType);
            }

            private async void ListenTimeout(Type protoType)
            {
                await CTask.Wait(5);
                if (task != null)
                {
                    Debug.LogError($"RPC 请求超时 [protoId:{protoId}] {protoType.Name}");
                    SetResult(null);
                    if (taskQueue.Count > 0) taskQueue.Dequeue();
                }
            }

            public void SetResult(object obj)
            {
                if (task != null)
                {
                    task.SetResult(obj);
                    task = null;
                }
            }
        }
        public int timestampOffsetS { get; private set; }

        public delegate void DisconnectAction(bool isActiveDisconnect);
        public event DisconnectAction onMainThreadDisconnect, onSubThreadDisconnect;
        public delegate void ReconnectAction(int times, bool success);
        public delegate void MsgAction(cs_msg_proto msg);
        public event ReconnectAction onMainThreadReconnect;
        public event MsgAction onMainThreadReceiveMsg, onSubThreadReceiveMsg;

        public string tag { private set; get; }
        private string key;
        private string ip;
        private int port;
        private bool isConnected;
        private Socket socket = null;
        private bool isReceiveRuning;
        private MsgPacker msgPacker;
        private Queue<Action> mainThread = new Queue<Action>();
        private bool isActiveDisconnect;
        private Dictionary<int, Queue<RPCTask>> rpcDic = new Dictionary<int, Queue<RPCTask>>();
        //监听网络断开
        private bool isListenDisconnect;
        private int disconnectNextTimeS;
        //异常断线自动重连
        private bool isReconnecting, hasTryReconnect;
        private int reconnectNextTimeS, reconnectTimes;

        //tag:用于多个组件时，拼接打印日志加以区分
        public void Awake(string serverKey, string tag)
        {
            key = AESHelper.DecryptString(serverKey);
            this.tag = tag;
        }

        public void Update()
        {
            while (mainThread.Count > 0)
            {
                Action action = mainThread.Dequeue();
                action?.Invoke();
            }
        }

        public bool IsConnected
        {
            get
            {
                if (!isConnected)
                {
                    return false;
                }
                return !((socket.Poll(1000, SelectMode.SelectRead) && (socket.Available == 0)) || !socket.Connected);
            }
        }

        public async CTask<bool> Connect(string ipAndPort)
        {
            string[] arr = ipAndPort.Split(':');
            return await Connect(arr[0], int.Parse(arr[1]));
        }

        public async CTask<bool> Connect(string ip, int port)
        {
            if (isConnected)
            {
                Log($"[{tag}] 当前已在连接状态");
                return true;
            }
            this.ip = ip;
            this.port = port;
            CTask<bool> task = CTask<bool>.Create();
            new Thread(() =>
            {
                try
                {
                    socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                    socket.Connect(ip, port);
                    bool success = !((socket.Poll(1000, SelectMode.SelectRead) && (socket.Available == 0)) || !socket.Connected);
                    if (success)
                    {
                        msgPacker = new MsgPacker();
                        Task.Run(ListenReceive);
                        Task.Run(ListenDisconnect);
                        isConnected = true;
                        isActiveDisconnect = false;
                        mainThread.Enqueue(() =>
                        {
                            task.SetResult(true);
                        });
                    }
                }
                catch (Exception ex)
                {
                    LogError($"[{tag}] 连接服务器失败: [{ip}:{port}] {ex}");
                    mainThread.Enqueue(() =>
                    {
                        task.SetResult(false);
                    });
                }
            }).Start();
            return await task;
        }

        private void ListenReceive()
        {
            isReceiveRuning = true;
            while (isReceiveRuning)
            {
                try
                {
                    byte[] buffer = new byte[8192];
                    int receiveLength = socket.Receive(buffer);
                    if (receiveLength > 0)
                    {
                        byte[] recBytes = new byte[receiveLength];
                        Array.Copy(buffer, 0, recBytes, 0, receiveLength);
                        msgPacker.OnReceiveBytes(recBytes, (byte[] bytes) =>
                        {
                            var msg = ProtobufHelper.ToObject<cs_msg_proto>(bytes);
                            onSubThreadReceiveMsg?.Invoke(msg);
                            mainThread.Enqueue(() =>
                            {
                                if (rpcDic.ContainsKey(msg.ProtoID))
                                {
                                    Queue<RPCTask> taskQueue = rpcDic[msg.ProtoID];
                                    if (taskQueue.Count > 0)
                                    {
                                        RPCTask rpcTask = taskQueue.Dequeue();
                                        object sc = ProtobufHelper.ToObject(msg.Body, rpcTask.task.GetResultType());
                                        rpcTask.SetResult(sc);
                                    }
                                }
                                onMainThreadReceiveMsg?.Invoke(msg);
                            });
                        });
                        continue;
                    }
                }
                catch { }
            }
        }

        //监听断开
        private void ListenDisconnect()
        {
            if (isListenDisconnect) return;

            int checkIntervalS = 2;
            Task.Run(() =>
            {
                isListenDisconnect = true;
                disconnectNextTimeS = TimeHelper.GetLocalTimestampS() + checkIntervalS;
                while (isListenDisconnect)
                {
                    Thread.Sleep(1000);
                    int nowS = TimeHelper.GetLocalTimestampS();
                    if (nowS >= disconnectNextTimeS)
                    {
                        disconnectNextTimeS = nowS + checkIntervalS;
                        if (!IsConnected)
                        {
                            isListenDisconnect = false;
                            if (isConnected)
                            {
                                if (isActiveDisconnect)
                                {
                                    Log($"[{tag}] 连接主动断开");
                                }
                                else
                                {
                                    LogError($"[{tag}] 连接被动断开");
                                }
                                Close();
                                onSubThreadDisconnect?.Invoke(isActiveDisconnect);
                                mainThread.Enqueue(() => { onMainThreadDisconnect?.Invoke(isActiveDisconnect); });
                                if (!isActiveDisconnect)
                                {
                                    mainThread.Enqueue(() =>
                                    {
                                        TryReconnect();
                                    });
                                }
                            }
                        }
                    }
                }
            });
        }

        //异常断开，启动重连
        private async void TryReconnect()
        {
            if (isReconnecting) return;

            Log($"[{tag}] 尝试重连");
            int checkIntervalS = 5;
            reconnectTimes = 1;
            isReconnecting = true;
            hasTryReconnect = false;
            reconnectNextTimeS = TimeHelper.GetLocalTimestampS() + checkIntervalS;
            while (isReconnecting)
            {
                await CTask.Wait(1);
                int nowS = TimeHelper.GetLocalTimestampS();
                if (nowS >= reconnectNextTimeS)
                {
                    reconnectNextTimeS = nowS + checkIntervalS;
                    if (isConnected)
                    {
                        if (hasTryReconnect)
                        {
                            mainThread.Enqueue(() =>
                            {
                                onMainThreadReconnect?.Invoke(reconnectTimes, true);
                            });
                        }
                        isReconnecting = false;
                        continue;
                    }

                    Log($"[{tag}] 重连");
                    hasTryReconnect = true;
                    ++reconnectTimes;
                    bool success = await Connect(ip, port);
                    if (success)
                    {
                        Log($"[{tag}] 重连成功");
                        isReconnecting = false;
                    }
                    mainThread.Enqueue(() =>
                    {
                        onMainThreadReconnect?.Invoke(reconnectTimes, success);
                    });
                }
            }
        }

        /// <summary>
        /// RPC方法
        /// </summary>
        public async CTask<T> RPC<T>(int csProtoId, int scProtoId, object csObj) where T : class
        {
            Queue<RPCTask> taskQueue = null;
            if (rpcDic.ContainsKey(scProtoId))
            {
                taskQueue = rpcDic[scProtoId];
            }
            else
            {
                taskQueue = new Queue<RPCTask>();
                rpcDic.Add(scProtoId, taskQueue);
            }
            CTask task = CTask.Create(typeof(T));
            RPCTask rpctask = new RPCTask(csProtoId, task, taskQueue, csObj.GetType());
            taskQueue.Enqueue(rpctask);
            Send(csProtoId, csObj);
            await task;
            return task.Result as T;
        }

        /// <summary>
        /// 发送消息
        /// </summary>
        public void Send(int protoId, object obj)
        {
            try
            {
                byte[] body = ProtobufHelper.ToBytes(obj);
                int timestamp = TimeHelper.GetLocalTimestampS() + timestampOffsetS;
                //拼接 body 的json只能用LitJson,与服务器统一，不能用Newtonsoft.json或者Unity自带的JsonUtility
                string sign = MD5Helper.StringToMD5(protoId.ToString() + body.ToJson() + key + timestamp);

                cs_msg_proto message = new cs_msg_proto()
                {
                    ProtoID = protoId,
                    Body = body,
                    TimeStamp = timestamp,
                    Sign = sign
                };

                byte[] msg = ProtobufHelper.ToBytes(message);
                byte[] msgWidthHead = MsgPacker.AddHead(msg);

                socket.Send(msgWidthHead);
            }
            catch (Exception e)
            {
                Log($"[{tag}] {e}");
            }
        }

        /// <summary>
        /// 设置服务器时间戳（秒）
        /// </summary>
        public void SetServerTimestamp(int serveTimestampS)
        {
            timestampOffsetS = serveTimestampS - TimeHelper.GetLocalTimestampS();
        }

        private void Close()
        {
            isReceiveRuning = false;
            isConnected = false;
            isListenDisconnect = false;
            foreach (var item in rpcDic.Values)
            {
                while (item.Count > 0)
                {
                    item.Dequeue().task.Dispose();
                }
            }
            rpcDic.Clear();
            if (socket != null)
            {
                try
                {
                    socket.Shutdown(SocketShutdown.Both);
                }
                catch { }
                try
                {
                    socket.Close();
                }
                catch { }
                socket = null;
            }
        }

        /// <summary>
        /// 主动断开连接
        /// </summary>
        public void Disconnect()
        {
            isActiveDisconnect = true;
            Close();
        }

        public void OnDestroy()
        {
            isReconnecting = false;
            Close();
        }
    }
}
